package com.yunhe.marketing.task;

import com.yunhe.marketing.domain.Enterprise;
import com.yunhe.marketing.esdomain.EsDataClueDemand;
import com.yunhe.marketing.esrepository.EsDataClueDemandrepository;
import com.yunhe.marketing.repository.EnterpriseRepository;
import com.yunhe.marketing.util.BeansUtil;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @ClassName SynchronizeDataController
 * @Description TODO
 * @Author HeTao
 * @Date 2021/5/26 16:08
 * @Version 1.0
 **/
@RestController
@RequestMapping("/synchronizes")
public class SynchronizeDataController {


    private static final Logger logger = LoggerFactory.getLogger(SynchronizeDataController.class);

    @Autowired
    private EnterpriseRepository enterpriseRepository;

    @Autowired
    private EsDataClueDemandrepository esDataClueDemandrepository;

    private static LinkedBlockingQueue<List<Enterprise>> dataListBOQueue = new LinkedBlockingQueue<>();

    private static int count = 100;

    private  int MAXSIZE = 1 ;

    @PostMapping("/synchronizeMysqlDataToEs")
    @Scheduled(cron = "0 0 1 * * ? ")
    @Async
    public void synchronizeMysqlDataToEs() throws InterruptedException {
        long startTime = System.currentTimeMillis();
        //创建线程池 批量处理数据
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(() -> {
            batchDataProcessing();
        });
        while (true){
            Thread.sleep(1000);
            List<Enterprise> enterprises = enterpriseRepository.getDataClueDemandBySynchronizeFlag(count);
            if(CollectionUtil.isNotEmpty(enterprises)){dataListBOQueue.add(enterprises);
            }else {
                break;
            }
        }
        executorService.shutdown();
        executorService.awaitTermination(20, TimeUnit.SECONDS);
        long endTime = System.currentTimeMillis();
        logger.info("全部数据处理所需时间："+(endTime-startTime));
    }

    private void batchDataProcessing() {
        try {
            for (; ; ) {
                List<Enterprise> dataLists = new ArrayList<>(100);
                for (int i = 0; i < MAXSIZE; i++) {
                    List<Enterprise> enterprises = dataListBOQueue.take();
                    if(CollectionUtil.isEmpty(enterprises)){
                        if(!CollectionUtil.isNotEmpty(dataLists)){
                            batchDataProcessing(dataLists);
                        }
                        return;
                    }else {
                        dataLists.addAll(enterprises);
                    }
                }
                if (!CollectionUtil.isEmpty(dataLists)) {
                    batchDataProcessing(dataLists);
                    dataLists.clear();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void batchDataProcessing(List<Enterprise> dataLists) {
        synchronizeDataToEs(enterpriseRepository,esDataClueDemandrepository,count,dataLists);
    }

    private void synchronizeDataToEs(EnterpriseRepository enterpriseRepository, EsDataClueDemandrepository esDataClueDemandrepository, int count, List<Enterprise> dataLists) {
        try {
            //CountDownLatch latch = new CountDownLatch(10);
            //ExecutorService executorSer = new ThreadPoolExecutor(10, 30, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(2000), new ThreadPoolExecutor.CallerRunsPolicy());
            if(CollectionUtil.isNotEmpty(dataLists)){
                long startTime1 = System.currentTimeMillis();
                List<EsDataClueDemand> esDataClueDemands = BeansUtil.listCopy(dataLists, EsDataClueDemand.class);
                esDataClueDemandrepository.saveAll(esDataClueDemands);
                enterpriseRepository.updateSynchronizeFlagById(esDataClueDemands);
                //executorSer.execute(new SynchronizeDataBatchRunnable(latch, enterpriseRepository, esDataClueDemandrepository, esDataClueDemands));
                long endTime1 = System.currentTimeMillis();
                logger.info(count+"条数据处理所需时间："+(endTime1-startTime1));
            }

            //latch.wait();
            //executorSer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
